#include "stream/SequenceReader.h"
#include <iostream>
#include "stream/Topic.h"

namespace stream {

SequenceReader::SequenceReader(std::shared_ptr<Topic> topic) 
: KvReader<key_type ,value_type>(topic)
{
}
SequenceReader::~SequenceReader(){
}

class MyKvReader : public SequenceReader {
public:

    MyKvReader(std::shared_ptr<Topic> topic, size_t offset)
    : SequenceReader(topic)
    , _offset(offset) 
    {
    }
	~MyKvReader() override {
		this->close();
    }
	
	bool hasNext() const override{
		return _offset < MAX;
    }
	SequenceReader::record_type current() const override{
        static std::shared_ptr<std::string> str(new std::string("Key-Value"));
		return std::make_pair(_offset, str);
    }
	void next() override{
		++_offset;
    }
protected:
	bool _open() override{
        return true;
	}
	void _close() override{
        _offset = 0;
	}
private:
    size_t _offset; 
    const size_t MAX = 3;
};



class TopicKvReader : public SequenceReader {
public:
	TopicKvReader(std::shared_ptr<Topic> topic, size_t offset)
	: SequenceReader(topic)
	, _offset(offset)
	{
	}
	~TopicKvReader() override{
		this->close();
	}

	bool hasNext() const override {
		return _offset < topic()->size();
	}
	SequenceReader::record_type current() const override {
		std::shared_ptr<std::string> value(new std::string);
		if(!topic()->get(std::to_string(_offset), *value)){
			std::cerr  << "get value from topic failed!" << std::endl;
		}
		return std::make_pair(_offset, value);
	}
	void next() override {
		++_offset;
	}

protected:
	bool _open() override{
		bool ret = false;
		if(!topic()->isOpened()){
			ret = topic()->open();
		}
		else // already opened
			ret = true;

		if(_offset >= topic()->size()){
			_offset = topic()->size() - 1;
		}

		return ret;
	}
	void _close() override{
		_offset = 0;
		if(topic()->isOpened()){
			if(topic().use_count() <= 1){
				topic()->close();
			}
		}
	}
private:
	size_t _offset = 0;
};

SequenceReader* makeSequenceReaderMock(std::shared_ptr<Topic> topic, size_t offset){
    return new MyKvReader(topic, offset);
}
SequenceReader * makeSequenceReader(std::shared_ptr<Topic> topic, size_t offset){
    return new TopicKvReader(topic, offset);
}

}//end namespace stream
